/*package org.spark.sample

import com.redis._
import spray.json._
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import spray.json.DefaultJsonProtocol
import com.redis.RedisClient
import scala.collection.mutable.ArrayBuffer

*//**
 * @author 325175
 *  WordCount例子扩展，统计完数据后保存到Redis上面，然后读取已有数据，合并历史和当前的统计结果
 *  这里会涉及到以下技术:
 *  1.Redis读写数据
 *  2.Json序列化和反序列化
 *//*
object WordCountAppWithRedis {

  def main(args: Array[String]) {
    var param = Array("local[2]", "D:/logs/doc.txt", "10.118.58.17", "6379","world_count")

    if (args.length != 0) {
      param = args
    }
    
    //1.先设置sparkConf对象
    val conf = new SparkConf().setAppName("WordCountAppWithRedis").setMaster(param(0))
    val sc = new SparkContext(conf)
    
    //2.加载文本
    val textFile = sc.textFile(param(1))
    
    val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => {
      val r = word.replaceAll(",","").replaceAll(":","")
      (r,1)
    }).reduceByKey(_+_)
    
    //3.加载已有的历史数据，这里涉及redis客户端使用和json转换
    
     //准备要转换的代码
    case class Word(word : String , count : Int)
    // 这个是转换器
    object MyJsonProtocol extends DefaultJsonProtocol {
      implicit val wordFormat = jsonFormat2(Word)
    }

    import MyJsonProtocol._
    
    // 读取已有的结果，进行转换,这里使用redis
    val r = new RedisClient(param(2), param(3).toInt)
    val savedJsons = r.get(param(4)) // 这里要json转对象

    println(savedJsons)
    // 这里使用模式匹配
    val saveResult = savedJsons match{
      case Some(s) => s.toString.parseJson.convertTo(DefaultJsonProtocol.listFormat[Word])
      case None => List[Word]()
      case _ => List[Word]()
    }
    
    //转换成RDD
    val savedCounts = sc.parallelize(saveResult).map(v => (v.word,v.count))
    
    val totalCounts = wordCounts.union(savedCounts).reduceByKey(_+_)
    
    //    totalCounts.collect().foreach(println _)


    // 保存结果到redis里面，先转换为List
    val resultList = ArrayBuffer[Word]()
    totalCounts.collect().foreach(r => resultList+= Word(r._1,r._2))

    println("===============================================")
    println(resultList.toArray.toJson)
    r.set(param(4),resultList.toArray.toJson)
    
  }
}*/